package hbase

import hbase.HBaseUtils.getTable
import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Delete, Get, Put, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes

import java.io.IOException


/**
 * Created with IntelliJ IDEA.
 *
 * @Auther: libobo
 * @Date: 2021/09/28/上午10:30
 * @Description:
 */
class HBaseUtils(val tableName: String) {
  val table = getTable(tableName)

  //向hbase表中插入数据
  //put 'sk:test1','1','i:name','Luck2'
  def insertTable(rowKey: String, columnFamily: String, column: String, value: String): Unit = {
    //准备key 的数据
    val puts = new Put(rowKey.getBytes())
    //添加列簇名,字段名,字段值value
    puts.addColumn(columnFamily.getBytes(), column.getBytes(), value.getBytes())
    //把数据插入到table中
    table.put(puts)
    println("insert data successful!!")
  }

  //获取hbase表中的数据
  //scan 'sk:test1'
  def scanDataFromHTable(columnFamily: String, column: String): Unit = {
    //定义scan对象
    val scan = new Scan()
    //添加列簇名称
    scan.addFamily(columnFamily.getBytes())
    //从table中抓取数据来scan
    val scanner = table.getScanner(scan)
    var result = scanner.next()
    //数据不为空时输出数据
    while (result != null) {
      println(s"rowKey:${Bytes.toString(result.getRow)},列簇:${columnFamily}:${column},value:${Bytes.toString(result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(column)))}")
      result = scanner.next()
    }
    //通过scan取完数据后，记得要关闭ResultScanner，否则RegionServer可能会出现问题(对应的Server资源无法释放)
    scanner.close()
  }

  //删除某条记录
  //delete 'sk:test1','1','i:name'
  def deleteRecord(rowKey: String, columnFamily: String, column: String): Unit = {
    val info = new Delete(Bytes.toBytes(rowKey))
    info.addColumn(columnFamily.getBytes(), column.getBytes())
    table.delete(info)
    println(s"delete rowKey:${rowKey},列簇:${columnFamily}:${column} successful!!")
  }
}

object HBaseUtils {
  val conf = HBaseConfiguration.create()
//  //设置zooKeeper集群地址，也可以通过将hbase-site.xml导入classpath，但是建议在程序里这样设置
//  conf.set("hbase.zookeeper.quorum", "dw-Huake05")
//  //设置zookeeper连接端口，默认2181
//  conf.set("hbase.zookeeper.property.clientPort", "2181")
//  //设置hbase连接端口，主机，默认60010
//  conf.set("hbase.master","dw-Huake05:60010")

  conf.set("hbase.zookeeper.quorum","master")
  conf.set("hbase.zookeeper.property.clientPort", "2181")
  conf.set("hbase.master","master:60010")

  val connection = ConnectionFactory.createConnection(conf)
  val admin = connection.getAdmin()
  //  val table = connection.getTable(TableName.valueOf("myTable"))

  def getTable(tableName: String): Table = {
    val tName = TableName.valueOf(tableName)
    if (!admin.tableExists(tName)) {
      val descriptor = new HTableDescriptor(tName)
      admin.createTable(descriptor)
      println("don't have the table, create table " + tableName + " successful")
    }
    var table = connection.getTable(tName)
    return table
  }

  def createTable(tableName: String, columnFamilies: Array[String]): Unit = {
    val tName = TableName.valueOf(tableName)
    if (!admin.tableExists(tName)) {
      val descriptor = new HTableDescriptor(tName)
      for (columnFamily <- columnFamilies) {
        descriptor.addFamily(new HColumnDescriptor(columnFamily))
      }
      admin.createTable(descriptor)
      println("create table " + tableName + " successful")
    }
  }

  def isTableExists(tableName: String): Boolean = {
    var result = false
    val tName = TableName.valueOf(tableName)
    if (admin.tableExists(tName)) {
      result = true
    }
    result
  }

  def dropTable(tableName: String): Unit = {
    admin.disableTable(TableName.valueOf(tableName))
    admin.deleteTable(TableName.valueOf(tableName))
    println("drop successful!!")
  }

  // 关闭 connection 连接
  def closeConnection(): Unit = {
    if (connection != null) {
      try {
        connection.close()
        println("关闭成功!")
      } catch {
        case e: IOException => println("关闭失败!")
      }
    }
  }
}